/******************************************
 * Author: Dave Spadea, Spadea Enterprises, Inc.
 * Date: November 2010
 * 
 * This code is released by Spadea Enterprises, Inc for general consumption 
 * under the Apache Software License.
 * 
 */


package net.spadea.framework.controller;

import net.spadea.framework.controller.logger.ControllerLoggerService;
import net.spadea.framework.messaging.*;
import net.spadea.framework.service.*;
import net.spadea.framework.persistence.*;

import java.sql.*;
import java.util.Calendar;
import java.util.Hashtable;

public class Controller extends FrameworkService {

	private static Hashtable<String, Object> routes;
	private static QueueChannel messageLog;
	private static Calendar nextRouteReloadTime; 
	
	/**
	 * @param args
	 * @throws ClassNotFoundException 
	 * @throws SQLException 
	 */
	public static void main(String[] args) throws SQLException, ClassNotFoundException {
		
		ControllerLoggerService cls = new ControllerLoggerService();
		cls.start_workers(50);
		log.info("Controller logger service workers started.");
		
		Controller c = new Controller();
		c.start_workers(50);
		log.info("Controller workers started.");
	}
	
	Controller() {
		super();

		routes = new Hashtable<String, Object>();
		nextRouteReloadTime = Calendar.getInstance();
		
		messageLog = new QueueChannel();
		messageLog.setName("controller.log");
		
		loadRouteInfo();
	}
	
	public void processMessage(Message msg) {
		log.info("Processing message ID: " + msg.getHeader("message_id"));
		
		synchronized(nextRouteReloadTime) {
			if ( nextRouteReloadTime.before(Calendar.getInstance()) ) {
				log.info("Route cache has expired. Reloading route info."); 
				loadRouteInfo();
			}
		}
		
		ChannelList channelList = null;
		synchronized (routes) {
			channelList = getRoutes(msg);
		}
		
		if (! channelList.isEmpty()) {
			for (Channel channel : channelList) {
				log.info("Dispatching to: " + channel.toString());
				msg.send(channel); // send message to the configured route
				logMessage(msg, channel);
			}
		} else {
			// No routes configured. Just log the message and move on.
			logMessage(msg, null);
		}
			
		
		
	}
	
	private void logMessage(Message m, Channel c) {
		// copy all messages to the logging service after setting route information headers
		if(c != null) {
			m.setHeader("routed_to", c.getName());
			m.setHeader("route_type", c.getType());
		} else {
			m.setHeader("routed_to", null);
			m.setHeader("route_type", null);
		}
		
		m.send(messageLog); 
	}
	
	@Override
	public Channel listenChannel() { return new QueueChannel("controller.request"); }
	
	private synchronized void loadRouteInfo() {
		
		log.info("Loading route info...");
		
		synchronized(routes) {
			synchronized(nextRouteReloadTime){
				log.info("Locks acquired.");
				
				routes = new Hashtable<String, Object>();
				nextRouteReloadTime = Calendar.getInstance();
				nextRouteReloadTime.add(Calendar.MINUTE, 1);
				
				try {
					Connection conn = Database.getNewConnection();
					Statement st = conn.createStatement();
					ResultSet rs = st.executeQuery("SELECT a.name as app_name, a.guid as app_guid, mt.name as type_name, " +
							"						       mr.status, mr.route_to, mr.route_channel_type " +
							"						FROM app a" +
							"						inner join message_type mt " +
							"							on a.id = mt.app_id" +
							"						inner join message_route mr " +
							"							on mt.id = mr.message_type_id ");
					
					while (rs.next()) {
						Hashtable<String, Object> row = Database.toHashtable(rs);
						log.info(row.toString());
						
						if ( ! routes.containsKey(row.get("app_guid")) ) {
							routes.put( (String)row.get("app_guid"), new Hashtable<String, Object>());
						}
						
						Hashtable<String, Object> msgTypes = (Hashtable<String,Object>)routes.get(row.get("app_guid"));
						
						if ( ! msgTypes.containsKey(row.get("type_name"))) {
							msgTypes.put((String)row.get("type_name"), new Hashtable<String, Object>());
						}
						
						Hashtable<String, Object> statuses = (Hashtable<String, Object>)msgTypes.get(row.get("type_name"));
						
						if ( ! statuses.containsKey(row.get("status"))) {
							statuses.put((String)row.get("status"), new ChannelList());
						}
						
						ChannelList status_routes = (ChannelList)statuses.get(row.get("status"));
						
						Channel c = new Channel();
						c.setType((String)row.get("route_channel_type"));
						c.setName((String)row.get("route_to"));
						
						status_routes.add(c);
						
					}
					
					rs.close();
					st.close();
				} catch (SQLException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
					System.exit(1);
				} catch (ClassNotFoundException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
					System.exit(2);
				}
				
				
			}
		}

	}
	
	private ChannelList getRoutes(Message msg) {
		
		String app_guid = msg.getApplicationID();
		String msgType = msg.getMessageType();
		String msgStatus = msg.getStatus();
		
		if ( app_guid == null || msgType == null || msgStatus == null) {
			log.error("Message header did not contain fields required for routing!\n" + msg.toString());
			return new ChannelList();
		}
		
		if (! routes.containsKey(app_guid)) {
			log.info("No route data for this application"); 
			return new ChannelList();
		}
		
		Hashtable<String, Object> msgTypes = (Hashtable<String,Object>)routes.get(app_guid);
		
		if (! msgTypes.containsKey(msgType)) {
			log.info("No route data for this message type"); 
			return new ChannelList();
		}
		
		Hashtable<String, Object> statuses = (Hashtable<String, Object>)msgTypes.get(msgType);
		
		if (! statuses.containsKey(msgStatus)) {
			log.info("No route data for this status");
			return new ChannelList();
		}
		
		ChannelList statusRoutes = (ChannelList) statuses.get(msgStatus);

		log.info("Returning channelList: " + statusRoutes.toString());
		return statusRoutes;
		
	}
	
}
